package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.core.consumequeue.ConsumeQueueConsumeHandler;
import cn.me.alphamq.broker.event.model.broker.ConsumeSuccessEvent;
import cn.me.alphamq.common.entity.req.ConsumeSuccessReq;
import cn.me.alphamq.common.entity.resp.ConsumeSuccessResp;
import cn.me.alphamq.common.enumeration.AckStatusEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;

/**
 * 消费成功事件的监听器 TODO 这个消息还没有同步给从 broker
 *
 * @author f
 */
public class ConsumeSuccessListener implements Listener<ConsumeSuccessEvent> {

    @Override
    public void onReceive(ConsumeSuccessEvent event) throws Exception {
        ConsumeSuccessReq consumeSuccessReq = event.getConsumeSuccessReq();
        String topic = consumeSuccessReq.getTopic();
        String consumerGroup = consumeSuccessReq.getConsumerGroup();
        Integer queueId = consumeSuccessReq.getQueueId();
        Integer batchMsgNum = consumeSuccessReq.getBatchMsgNum();

        // TODO 有必要很仔细地校验 topic 是否存在，以及消费者实例是否被剔除吗？详见 7-17

        ConsumeQueueConsumeHandler consumeQueueConsumeHandler = Container.getConsumeQueueConsumeHandler();
        for (int i = 0; i < batchMsgNum; i++) {
            if (consumeQueueConsumeHandler.ack(topic, consumerGroup, queueId)) {
                continue;
            }
            // 更新 offset 失败，返回 ack 失败
            event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.CONSUME_SUCCESS_RESP, JSON.toJSONBytes(
                    new ConsumeSuccessResp(consumeSuccessReq.getMsgId(), AckStatusEnum.FAIL.getCode()))));
            return;
        }

        event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.CONSUME_SUCCESS_RESP, JSON.toJSONBytes(
                new ConsumeSuccessResp(consumeSuccessReq.getMsgId(), AckStatusEnum.SUCCESS.getCode()))));
    }
}
